package com.atguigu.tingshu.album.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;
import java.util.Set;

@CanalEventListener         // 声明当前这个类是Canal的事件监听器
@Slf4j
public class CanalClientListener {

    @Autowired
    private RedisTemplate<String , String> redisTemplate ;

    @Autowired
    private KafkaService kafkaService ;

    /**
     * 监听album_info表的数据变化
     */
    @ListenPoint(destination = "example" , schema = "tingshu_album" , table = "album_info" , eventType = { CanalEntry.EventType.UPDATE , CanalEntry.EventType.DELETE })
    public void albumInfoListener(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {

        // 获取修改之前数据
//        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();     // 获取修改之前的数据列表的列表
//        beforeColumnsList.forEach(column -> {
//            String columnName = column.getName();
//            String value = column.getValue();
//            System.out.println(columnName + ": " + value);
//        });

//        System.out.println("===============================================================================================");

        // 获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            String value = column.getValue();
            if("id".equals(columnName)) {
                redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX + value) ;
                log.info("删除了Redis中的专辑数据，albumId: {}" , value);
            }
        });

    }

    // 同步分类数据
    @ListenPoint(destination = "example" , schema = "tingshu_album" , table = { "base_category1" , "base_category2" , "base_category3" } , eventType = { CanalEntry.EventType.UPDATE , CanalEntry.EventType.DELETE })
    public void baseCategoryListener(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {

        // 只要分类数据修改了，就删除Redis中的分类数据
        String redisKey = RedisConstant.BASE_CATEGORY_VIEW_PREFIX + "*" ;
        Set<String> set = redisTemplate.keys(redisKey);
        set.forEach(key -> redisTemplate.delete(key));
        log.info("删除了Redis中所对应分类数据...");

    }

    // 同步用户
    @ListenPoint(destination = "example" , schema = "tingshu_user" , table = { "user_info" } , eventType = { CanalEntry.EventType.UPDATE , CanalEntry.EventType.DELETE })
    public void userInfoListener(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {

        // 获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            String value = column.getValue();
            if("id".equals(columnName)) {
                redisTemplate.delete(RedisConstant.USER_INFO_PREFIX + value) ;
                log.info("删除了Redis中的用户数据，userId: {}" , value);
            }
        });
    }

    // 同步专辑的统计数据
    @ListenPoint(destination = "example" , schema = "tingshu_album" , table = { "album_stat" } , eventType = { CanalEntry.EventType.UPDATE , CanalEntry.EventType.DELETE })
    public void albumStatListener(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {

        // 创建AlbumStat对象
        AlbumStat albumStat = new AlbumStat() ;

        // 获取修改之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach(column -> {
            String columnName = column.getName();
            String value = column.getValue();
            if("album_id".equals(columnName)) {
                redisTemplate.delete(RedisConstant.ALBUM_STAT_PREFIX + value) ;
                log.info("删除了Redis中的专辑的统计数据，albumId: {}" , value);
                albumStat.setAlbumId(Long.parseLong(value));
            }else if("stat_type".equals(columnName)) {
                albumStat.setStatType(value);
            }else if("stat_num".equals(columnName)) {
                albumStat.setStatNum(Integer.parseInt(value));
            }
        });

        // 修改ES索引库中的统计数据
        /**
         * 发送消息的内容：1、专辑的id   2、专辑的统计类型（0401-播放量 0402-订阅量 0403-购买量 0404-评论数） 3、专辑的统计数据
         */
        kafkaService.sendMsg(KafkaConstant.QUEUE_SYNC_ES_ALBUM_STAT , JSON.toJSONString(albumStat));

    }

}
